package cn.azzhu.day06;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

/**
 * @author azzhu
 * @create 2020-09-20 10:39:07
 */
public class KafkaSourceV2 {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //开启CheckPointing，同时开启重启策略
        env.enableCheckpointing(5000);
        //设置StateBackend
        env.setStateBackend(new FsStateBackend("file:///D:\\bigdata\\flink-learning\\backend"));
        //取消任务，checkpoint不删除
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //设置checkpoint模式
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        //kafkaSource
        final Properties props = new Properties();
        //指定kafka的broker地址
        props.setProperty("bootstrap.servers", "hadoop105:9092,hadoop106:9092,hadoop107:9092");
        //指定组ID
        props.setProperty("group.id", "ga10");
        //如果没有指定偏移量，第一次从最开始消费
        props.setProperty("auto.offset.reset","earliest");
        //kafka的消费者不自动提交偏移量
        props.setProperty("enable.auto.commit","false");

        final FlinkKafkaConsumer<String> kafkaSource = new FlinkKafkaConsumer<>(
                "activity10",
                new SimpleStringSchema(),
                props
        );

        //Flink checkpointing成功后，还要向kafka特殊的topic写入偏移量
        //这个参数默认是true，可以改为false，不建议修改
        kafkaSource.setCommitOffsetsOnCheckpoints(false);

        final DataStreamSource<String> words = env.socketTextStream("hadoop105", 9999);
        words.map(new MapFunction<String, String>() {
            @Override
            public String map(String line) throws Exception {
                if(line.startsWith("azzhu")) {
                    System.out.println(1/0);
                }
                return line;
            }
        }).print();

        //source
        final DataStream<String> lines = env.addSource(kafkaSource);

        //sink
        lines.print();

        env.execute("KafkaSourceV2");
    }
}
